package com.we.flink.cdcpipeline.test;

import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.SchemaBuilder;
import io.debezium.spi.converter.CustomConverter;
import io.debezium.spi.converter.RelationalColumn;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Properties;

public class MySQLTimeCustomConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {

    SimpleDateFormat sdf = new SimpleDateFormat("yyy-MM-dd HH:mm:ss");
    DateTimeFormatter datetimeFormatter = DateTimeFormatter.ISO_DATE_TIME;
    ZoneId timestampZoneId = ZoneId.systemDefault();

    @Override
    public void configure(Properties props) {}

    @Override
    public void converterFor(
            RelationalColumn column, ConverterRegistration<SchemaBuilder> registration) {

        String sqlType = column.typeName().toUpperCase();
        SchemaBuilder schemaBuilder = null;
        Converter converter = null;

        if ("DATETIME".equals(sqlType)) {
            schemaBuilder =
                    SchemaBuilder.string().optional().name("com.wetech.debezium.datetime.string");
            converter = this::convertDatetime;
        }

        if ("TIMESTAMP".equals(sqlType)) {
            schemaBuilder =
                    SchemaBuilder.string().optional().name("com.wetech.debezium.timestamp.string");
            converter = this::convertTimestamp;
        }

        if ("FLOAT".equals(sqlType)) {
            schemaBuilder =
                    SchemaBuilder.string().optional().name("com.wetech.debezium.float.string");
            converter = this::convertFloat;
        }

        if (schemaBuilder != null) {
            registration.register(schemaBuilder, converter);
        }
    }

    private String convertDatetime(Object input) {
        if (input == null) return null;
        if (input instanceof Timestamp) {
            return sdf.format((Timestamp) input);
        }
        // datatime datatype upsert
        if (input instanceof LocalDateTime) {
            return datetimeFormatter.format((LocalDateTime) input).replaceAll("T", " ");
        }
        return String.valueOf(input);
    }

    private String convertTimestamp(Object input) {
        if (input == null) return null;
        if (input instanceof Timestamp) {
            return sdf.format((Timestamp) input);
        }
        // timestamp datatype upsert
        if (input instanceof ZonedDateTime) {
            ZonedDateTime zonedDateTime = (ZonedDateTime) input;
            LocalDateTime localDateTime =
                    zonedDateTime.withZoneSameInstant(timestampZoneId).toLocalDateTime();
            return datetimeFormatter.format(localDateTime).replaceAll("T", " ");
        }
        return String.valueOf(input);
    }

    private String convertFloat(Object input) {
        if (input == null) return null;
        return String.valueOf(input);
    }
}
